package com.yl.flink.filter;

import cn.hutool.core.date.DateUtil;
import com.yl.constant.TimeUnit;
import com.yl.entity.MultiDataEntity;
import com.yl.util.DtUtil;
import org.apache.flink.api.common.functions.FilterFunction;

/**
 * @author wlf
 * @since 2022/8/18
 */
public class DataFilter implements FilterFunction<MultiDataEntity> {

    // 是否是历史数据
    private final Boolean hisFlag;
    // 历史数据延迟小时
    private final Integer latencyHour;

    public DataFilter(Boolean hisFlag, Integer latencyHour) {
        this.hisFlag = hisFlag;
        this.latencyHour = latencyHour;
    }

    /**
     * 对数据流进行过滤，筛选实时数据流或者历史数据流
     * 【过滤逻辑】
     * 判断数据时间和当前时间相差是否大于5小时，小于5小时的由influxdb的连续查询处理
     *
     * @param multiDataEntity 流元素
     */
    @Override
    public boolean filter(MultiDataEntity multiDataEntity) throws Exception {
        // 数据时间和当前时间相差的小时
        long diffTime = DtUtil.diffTime(DateUtil.date(), DateUtil.date(multiDataEntity.getPickTime() / 1000), TimeUnit.H);
        if (hisFlag) {
            // 历史数据流
            return diffTime >= latencyHour;
        } else {
            // 实时数据流
            return diffTime < latencyHour;
        }
    }
}
